package xin.alum.aim.redis;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import xin.alum.aim.cluster.ClusterPusher;
import xin.alum.aim.constant.AimConstant;
import xin.alum.aim.constant.Constants;
import xin.alum.aim.groups.ClusterFactory;
import xin.alum.aim.model.Pusher;


class RedisPusher extends ClusterPusher {

    @Autowired
    @Qualifier("aimRedisTemplate")
    private RedisTemplate<String, Object> redisTemplate;

    protected <T> boolean push(String channel, T d) {
        redisTemplate.convertAndSend(channel, d);
        return true;
    }

    /**
     * 积压消息处理
     */
    @Async
    @Override
    protected void pushQueue() {
        redisTemplate.executePipelined((RedisCallback<Object>) c -> {
            if (queue.size() > 10)
                logger.warn("集群消息出现积压:{}/{}", queue.size(), ClusterFactory.QUEUE_CAPACITY);
            Pusher m;
            while (queue.iterator().hasNext()) {
                m = queue.poll();
                if (m == null) break;
                switch (m.getData().getTag()) {
                    case AimConstant.DATA_TYPE_MESSAGE:
                        push(Constants.PUSH_MESSAGE_INNER_QUEUE, m);
                        break;
                    default:
                        push(Constants.EVENT_MESSAGE_INNER_QUEUE, m);
                        break;
                }
            }
            return null;
        });
    }
}
